#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <stdlib.h>
#include <sys/eventfd.h>
#include <pthread.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/time.h>

#include "catimer.h"
#include "caevent.h"
#include "calog.h"

extern void free_estimer_all(es_t* psys);

static void es_all_dereg(es_t* psys);
static void es_event_dereg(es_t* psys, evt_t* pevt);

static void open_msg_proc(es_t  *psys);
static void close_msg_proc(es_t *psys);

static void open_extra_ipc(es_t  *psys);
static void close_extra_ipc(es_t  *psys);

static evt_t* alloc_evt(es_t* psys);
static void free_evt(evt_t* pevt);

static void rfd_cb(evt_t* pevt, int fd, int events);

static int default_msg_proc(es_t* psys, uint16_t msg, uint16_t p1, uint32_t p2);


int es_open(es_t* psys)
{
	memset(psys, 0, sizeof(es_t));
	gettimeofday(&psys->tm, NULL);
	
	psys->epfd = epoll_create(10);
	psys->exit_flag = 0;
	psys->evt_count = 0;
	psys->pevtlist = NULL;
	psys->timer_list = NULL;
	
	psys->handle_seed = rand();

	
	open_msg_proc(psys);
	open_extra_ipc( psys);
	psys->opened = 1;
	return psys->epfd;
}

void es_close(es_t* psys)
{
	//logprt("=== evsys close\n");

	
	psys->msg_proc(psys, EM_CLOSE, 0, 0);
	
	free_estimer_all(psys);
	
	close_extra_ipc(psys);
	close_msg_proc( psys); // msg proc event releasing.....
	
	if(psys->pevtlist)
	{
		logprt("###   pending event close.........count=%d, \n", psys->evt_count);
		/*
		lklist* pevt = psys->pevtlist;
		do {
//			pnext = pevt->next;
			es_dereg(psys, (evt_t*)pevt);
//			pevt = pnext;
		} while(pevt != psys->pevtlist);
		*/

		es_all_dereg(psys);

	}
	
	close(psys->epfd);

	// check free resource
	if(psys->evt_alloc_cnt >0 )
	{
		logprt("### Error: event leak....\n");
		assert(0);
	}
	
	if(psys->timer_alloc_cnt >0 )
	{
		logprt("### Error: timer leak....\n");
		assert(0);
	}

	if(psys->timer_cnt > 0)
	{
		logprt("### Error: es timer leak....cnt=%d\n", psys->timer_cnt);
		assert(0);
	}

	if(psys->user)
	{
		free(psys->user);
	}
	psys->opened = 0;
}


int es_main(es_t* psys)
{
	int i;
	int nfds;
	struct epoll_event events[MAX_GET_EVENTS];
	struct epoll_event *epv;

	psys->msg_proc(psys, EM_INIT,  psys->init_p1,  psys->init_p2);

	for(;psys->exit_flag==0;)
	{
		nfds = epoll_wait(psys->epfd, events, MAX_GET_EVENTS, -1);
		//logprt("nfds = %d \n", nfds);
		struct epoll_event *epv;
		for(i=0;i<nfds;i++)
		{
			epv = events+i;
			evt_t* pevt = (evt_t*)epv->data.ptr;
			pevt->evtcb(pevt, pevt->fd, epv->events);
		}
	}

	return 0;
}

void es_ready(es_t* psys)
{
	for(;!psys->opened;)
	{
		logprt("==  open wait.....\n");
		usleep(1000);
	}
	es_sendmsg(psys, EM_READY, 0, 0);
}

static int default_msg_proc(es_t* psys, uint16_t msg, uint16_t p1, uint32_t p2)
{
	return 0;
}


int es_alloc_user(es_t* psys, int size)
{
	if(psys->user)
		return -1;
	psys->user = malloc(size);
	return 0;
}


void es_set_init_para(es_t* psys, uint16_t p1, uint32_t p2)
{
	psys->init_p1 = p1;
	psys->init_p2 = p2;
}


void es_register_msgproc(es_t* psys, EVTMSG_PROC proc)
{
	psys->msg_proc = proc;
}


void es_exit(es_t* psys)
{
	//char cmd;
	//write(psys->lfd, &cmd, 1);
	es_postmsg( psys, EM_EXIT, 0, 0);
}


evt_t* es_reg(es_t* psys, int fd, uint32_t events, EVENTCB cb, void* user)
{
	struct epoll_event event;
	if(cb==NULL)
		assert(0);
		
	evt_t *pevt = alloc_evt(psys);
	pevt->fd = fd;
	pevt->wevents = events;
	pevt->evtcb = cb;
	pevt->user = user;

	event.data.ptr = pevt;
	event.events = events;

	int ret = epoll_ctl(psys->epfd, EPOLL_CTL_ADD, fd, &event);

	// event list add
	if(psys->pevtlist==NULL)
	{
		lkinit(&pevt->list);
	}
	else
	{
		lkadd(psys->pevtlist, &pevt->list);
	}
	psys->pevtlist = &pevt->list;
	psys->evt_count++;
	
	
	//logprt("== reg event, fd=%d, event_obj=%p, event=%d, count=%d, ret=%d \n", fd, pevt, events, psys->evt_count, ret);
	
	return pevt;
}

void es_dereg(es_t* psys, evt_t* pevt)
{
	es_event_dereg(psys, pevt);
	psys->pevtlist = lkpop_head(&pevt->list);
	psys->evt_count--;
	//logprt("== dereg event, event_obj=%p, count=%d\n", pevt, psys->evt_count);	
	free_evt(pevt);
}

static void es_event_dereg(es_t* psys, evt_t* pevt)
{
	assert(pevt);
	if(pevt)
	{
		epoll_ctl(psys->epfd, EPOLL_CTL_DEL, pevt->fd, NULL);
		assert(psys->pevtlist);
	}		

}




static void es_all_dereg(es_t* psys)
{
	evt_t* pevt = (evt_t*)psys->pevtlist;
	for(;pevt;)
	{
		es_event_dereg(psys, pevt);
		psys->evt_count--;
		psys->pevtlist = lkpop_head(&pevt->list);
		free_evt(pevt);
		pevt = (evt_t*)psys->pevtlist;			
	}
}


int es_change(evt_t* pevt, uint32_t event)
{
	int ret;
	struct epoll_event ev;
	ev.events = event;
	ev.data.ptr = pevt;
	ret = epoll_ctl(pevt->psys->epfd, EPOLL_CTL_MOD, pevt->fd, &ev);
	logprt("== es_change, fd=%d, cb=%p, user=%p, ret=%d\n", pevt->fd, pevt->evtcb, pevt->user, ret);
	return ret;
}


static evt_t* alloc_evt(es_t* psys)
{
	static int handle=0;
	evt_t* pevt = (evt_t*)malloc(sizeof(evt_t));
	
	if(pevt)
	{
		memset(pevt, 0, sizeof(evt_t));
		pevt->psys = psys;
		psys->evt_alloc_cnt++;

		pevt->handle = handle++;
	}
	
	return pevt;
}

static void free_evt(evt_t* pevt)
{
	pevt->psys->evt_alloc_cnt--;
	free(pevt);
}




// ipc for long data
static void open_extra_ipc(es_t* psys)
{
	int sv[2];
	socketpair(AF_UNIX, SOCK_STREAM, 0, sv);
	psys->lfd = sv[0];
	psys->rfd = sv[1];
	psys->pevt_ipcext = es_reg(psys, psys->rfd, EPOLLIN, rfd_cb, psys);
//	logprt("=== open extra ipc,  fd = %d, evt=%p\n", psys->rfd, psys->pevt_ipcext);
}

static void close_extra_ipc(es_t* psys)
{
	//logprt("=== close extra ipc, fd = %d, evt=%p\n", psys->rfd, psys->pevt_ipcext);
	close(psys->lfd);
	close(psys->rfd);
	if(psys->pevt_ipcext)
		es_dereg(psys, psys->pevt_ipcext);
	psys->pevt_ipcext = NULL;	
}

void es_register_ipcproc(es_t* psys, EVTIPC_PROC proc)
{
	assert(!psys->ipc_proc);
	psys->ipc_proc = proc;
}


int es_send_ipc(es_t* psys, char* buf, int size)
{
	return write(psys->lfd, buf, size);
}

int es_read_ipc(es_t* psys, char* buf, int size)
{
	return read(psys->rfd, buf, size);
}



//........


//.. ipc message functions
static void evt_msg_cb(evt_t* pevt, int fd, int events)
{
	int ret;
	evtmsg_t msg;
	uint16_t msgid;
	ret = read(fd, &msg, sizeof(msg));
	msgid = (msg.msgid & 0x4fff);
	if(msgid==EM_EXIT) 
	{
		pevt->psys->exit_flag = 1;
	}
	else if(msgid == EM_READY)
	{
		logprt("== ES ready.... es=%p\n", pevt->psys);
	}
	else
		pevt->psys->msg_proc(pevt->psys, msgid, msg.p1, msg.p2);
	if(msg.msgid & 0x8000)
	{
		logprt("== send msg recv....id=%d\n", msgid);
		pthread_mutex_lock(&pevt->psys->msg_sync_mutex);
		pthread_cond_signal(&pevt->psys->msg_sig);
		pthread_mutex_unlock(&pevt->psys->msg_sync_mutex);
		logprt("== send msg recv end....id=%d\n", msgid);
		
	}
}

static void open_msg_proc(es_t  *psys)
{
	int ret;
	//logprt("open msg proc....\n");
	psys->msg_sync_mutex = PTHREAD_MUTEX_INITIALIZER;
	ret = pthread_mutex_init(&psys->msg_sync_mutex, NULL);
	//logprt("%s, ret=%d \n", __func__, ret);
	ret = pthread_cond_init(&psys->msg_sig, NULL);
	//logprt("%s, ret=%d \n", __func__, ret);
	psys->event_fd = eventfd(0, EFD_NONBLOCK);
	psys->pevt_msg = es_reg(psys,  psys->event_fd,  EPOLLIN,   evt_msg_cb, NULL);
	psys->msg_proc = default_msg_proc;
}

static void close_msg_proc(es_t *psys)
{
	close(psys->event_fd);
	es_dereg(psys, psys->pevt_msg);
	pthread_cond_destroy(&psys->msg_sig);
	pthread_mutex_destroy(&psys->msg_sync_mutex);
	
}



void es_postmsg(es_t* psys, uint16_t msgid, uint16_t p1, uint32_t p2)
{
	evtmsg_t msg;
	assert(!(msgid & 0x8000));
	msg.msgid = msgid;
	msg.p1 = p1;
	msg.p2 = p2;
	write(psys->event_fd, &msg, sizeof(msg)); // event fd system allow for just 8-byte msg.
}


int es_sendmsg(es_t* psys, uint16_t msgid, uint16_t p1, uint32_t p2, int *presult)
{
	evtmsg_t msg;
	assert(!(msgid & 0x8000));	
	msg.msgid = msgid | 0x8000;
	msg.p1 = p1;
	msg.p2 = p2;
	logprt(">> send enter mutex...\n");
	pthread_mutex_lock(&psys->msg_sync_mutex);
	write(psys->event_fd, &msg, sizeof(msg)); // event fd system allow for just 8-byte msg.
	pthread_cond_wait(&psys->msg_sig, &psys->msg_sync_mutex);
	if(presult)
		*presult = psys->mresult;
	pthread_mutex_unlock(&psys->msg_sync_mutex);
	logprt(">> send out mutex...\n");
	return 0;
}


unsigned int es_mtime(es_t *psys)
{
	struct timeval tm;
	gettimeofday(&tm, NULL);
	return  ( (tm.tv_sec-psys->tm.tv_sec)*1000 + (tm.tv_usec-psys->tm.tv_usec)/1000  );
}


void es_set_msg_result(es_t* psys, int result)
{
	psys->mresult = result;
}

// ==========================



static void rfd_cb(evt_t* pevt, int fd, int events)
{
	es_t* psys = (es_t*)pevt->psys;
	//assert(psys->ipc_proc);
	if(psys->ipc_proc)
		psys->ipc_proc(pevt->psys, fd);
}


void setNonBlockMode(int fd)
{
	int flags = fcntl(fd, F_GETFL);
	flags |= O_NONBLOCK;
	fcntl(fd, F_SETFL, flags);
}

